package com.boot.study.table

import com.boot.study.api.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row

object TimeAndWindowTest {
  def main(args: Array[String]): Unit = {
    // 1: 创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // TimeCharacteristic.EventTime 事件时间
    // TimeCharacteristic.ProcessingTime 处理时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 时间语义，处理时间

    // 创建表执行环境
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    val inputPath: String = "D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt"
    val inputSteam: DataStream[String] = env.readTextFile(inputPath)
    val dataStream: DataStream[SensorReading] = inputSteam.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
      // 延迟1秒生成 watermark
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = element.timeStamp * 1000
      })

    // 定义处理时间
    // val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timeStamp, 'pt.proctime)

    // 定义事件时间
    // 指定时间字段
    // val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timeStamp.rowtime, 'pt.rowtime)
    // 追加时间字段
    //  val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timeStamp, 'pt.rowtime)
    val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timeStamp.rowtime as 'ts)

    // sensorTable.printSchema()
    // sensorTable.toAppendStream[Row].print("row")

    // 1: group window
    val resultTable: Table = sensorTable
      // .window(Tumble.over(10.seconds).on('ts).as('w)) // 10秒的滚动窗口
      .window(Tumble over 10.seconds on 'ts as 'w) // 单一参数传递可以省略 点和括号，用空格隔开
      .groupBy('id, 'w) // 根据id分组
      .select('id, 'id.count, 'temperature.avg, 'w.end) // 'w.end 关窗时间

    // 2: sql window
    tableEnv.createTemporaryView("sensor", sensorTable)
    env.execute("time and window test")
    val resultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select
        |   id,
        |   count(id),
        |   avg(temperature),
        |   tumble_end(ts, interval '10' second)
        |from sensor
        |group by
        |id,
        |tumble(ts, interval '10' second)
        |""".stripMargin)

    // 2 over window:统计每个sensor与之前两行数据的平均值
    // 2.1 table
    val overResultTable: Table = sensorTable
      .window(Over partitionBy 'id orderBy 'ts preceding 2.rows as 'ow)
      .select('id, 'ts, 'id.count over 'ow, 'temperature.avg over 'ow)

    val overResultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select
        | id,
        | ts,
        | count(id) over ow,
        | avg(temperature) over ow
        | from sensor
        | window ow as (
        | partition By id
        | order by ts
        | rows between 2 preceding and current row
        | )
        |""".stripMargin)

    // 3: 转换流输出
//    resultTable.toAppendStream[Row].print("table")
//    resultSqlTable.toRetractStream[Row].print("sql")

    overResultTable.toAppendStream[Row].print("over table")
    overResultSqlTable.toRetractStream[Row].print("over sql")

    env.execute("time and window test")
  }
}
